#include "znx_chash.h"
#include "znx_random.h"


znx_chash_t *
znx_chash_create(size_t replace)
{
    znx_chash_t     *chash;

    if (replace < 1) {
        replace = ZNX_CHASH_DEF_REPLACE;
    }

    chash = znx_thread_calloc(sizeof(znx_chash_t));
    chash->replace = replace;
    chash->nodes = znx_array_create(8, sizeof(znx_chash_node_t));

    return chash;
}


void
znx_chash_add_node(znx_chash_t *chash, znx_str_t *key, void *data, size_t weight)
{
    znx_chash_node_t        *node;

    if (weight <= 0) {
        return;
    }

    node = znx_array_push(chash->nodes);
    znx_chash_node_init(node, key, data, weight);
}


void
znx_chash_build(znx_chash_t *chash)
{
    ZNX_ASSERT_FATAL(chash->vnodes);

    znx_chash_node_t    *node, *nodes;
    znx_chash_vnode_t   *vnode;
    size_t              i, j;
    size_t              nvondes = 0;

    if (chash->nodes->nelts == 0) {
        return;
    }
    
    nodes = chash->nodes->elts;
    for (i = 0; i < chash->nodes->nelts; i++) {
        node = nodes + i;
        chash->total_weight += node->weight;
        nvondes += (chash->replace * node->weight);
    }

    chash->nvnodes = nvondes;
    chash->vnodes = znx_thread_malloc(nvondes * sizeof(znx_chash_vnode_t));
    vnode = chash->vnodes;

    for (i = 0; i < chash->nodes->nelts; i++) {
        node = nodes + i;
        for (j = 0; j < chash->replace; j++) {
            vnode = (vnode + 1);
            znx_chash_vnode_init(vnode, node, (int)j);
        }
    }

    znx_qsort(
        chash->vnodes,
        chash->nvnodes,
        sizeof(znx_chash_vnode_t),
        znx_chash_vnode_cmp_handle);
}


static void *
znx_chash_get_by_random(znx_chash_t *chash)
{
    size_t              cur_weight = 0;
    size_t              i;
    znx_chash_node_t    *node, *nodes;
    size_t              rand;
    
    rand = (size_t)znx_random_n((int)chash->total_weight);

    nodes = chash->nodes->elts;

    for (i = 0; i < chash->nodes->nelts; i++) {
        node = nodes + i;
        cur_weight += node->weight;
        if (rand < cur_weight) {
            return node->data;
        }
    }

    ZNX_ASSERT_FATAL("unreachable");
}


static void *
znx_chash_get_by_random_and_checker(znx_chash_t *chash,
    znx_chash_node_check_pt checker, void *user_data)
{
    size_t              total = 0;
    size_t              cur = 0;
    size_t              i, rand;
    znx_chash_node_t    *node, *nodes;

    u_char *status = znx_thread_calloc(sizeof(u_char) * chash->nodes->nelts);

    nodes = chash->nodes->elts;

    for (i = 0; i < chash->nodes->nelts; i++) {
        node = nodes + i;
        if (checker && checker(node, user_data)) {
            total += node->weight;
            status[i] = 1;
        }
    }

    rand = (size_t)znx_random_n((int)total);

    for (i = 0; i < chash->nodes->nelts; i++) {
        if (!status[i]) {
            continue;
        }

        node = nodes + i;
        cur += node->weight;
        if (rand < cur) {
            znx_thread_free(status);
            return node->data;
        }
    }

    ZNX_ASSERT_FATAL("unreachable");
}


static uint32_t
znx_chash_get_server_index(znx_chash_t *chash, uint32_t hash)
{
    uint32_t    low, hight, mid, crc_key;
    low = 0;
    hight = (uint32_t)chash->nvnodes;

    while (low < hight) {
        mid = (low + hight) >> 1;
        crc_key = (chash->vnodes + mid)->crc_key;

        if (crc_key == hash) {
            return mid;
        } else if (crc_key < hash) {
            low = mid + 1;
        } else {
            hight = mid;
        }
    }

    if (low == chash->nvnodes && (chash->vnodes + low)->crc_key < hash) {
        return 0;
    }

    return low;
}


// if balance < 0, it means that want force_balance.

znx_chash_result_t
znx_chash_get(znx_chash_t *chash,
    znx_str_t *key, int balance,
    znx_chash_node_check_pt checker,
    void *user_data,
    znx_bool_t must_get)
{
    znx_chash_result_t  result;
    bzero(&result, sizeof(znx_chash_result_t));

    znx_chash_node_t    *node;
    znx_chash_vnode_t   *vnode;

    if (chash->vnodes == NULL) {
        return result;
    }

    if (chash->nodes->nelts == 1) {
        node = chash->nodes->elts;
        if (checker && !checker(node, user_data)) {
            if (must_get) {
                result.server = node->data;
            }
            return result;
        }

        result.server = node->data;
        return result;
    }

    uint32_t            crc;
    size_t              real_index;
    size_t              i;

    crc = znx_crc32_long(key->data, key->len);
    real_index = znx_chash_get_server_index(chash, crc);

    for (i = 0; i < chash->nvnodes; i++) {
        real_index = (real_index + i) % chash->nvnodes;
        vnode = chash->vnodes + real_index;
        if (checker && !checker(vnode->real_node, user_data)) {
            continue;
        }

        result.server = vnode->real_node->data;
        break;
    }

    if (result.server == NULL) {
        if (must_get) {
            result.server = znx_chash_get_by_random(chash);
        }
        return result;
    }

    if (balance < 0) {
        result.balance_server = znx_chash_get_by_random_and_checker(chash, checker, user_data);
        return result;
    }

    if (balance <= 1) {
        return result;
    }

    size_t              select_index;
    size_t              index;
    select_index = (size_t)(znx_random_n(10000) % balance);

    for (i = 0; i < chash->nvnodes; i++) {
        index = (real_index + select_index) % chash->nvnodes;
        vnode = chash->vnodes + index;
        if (checker && !checker(vnode->real_node, user_data)) {
            continue;
        }

        result.balance_server = vnode->real_node->data;

        if (result.balance_server == result.server) {
            result.balance_server = NULL;
        }

        break;
    }

    return result;
}


void
znx_chash_destroy(znx_chash_t *chash)
{
    if (chash->nodes != NULL) {
        znx_array_destroy(chash->nodes);
    }

    if (chash->vnodes != NULL) {
        znx_thread_free(chash->vnodes);
    }

    znx_thread_free(chash);
}

